package distributed

import (
	"context"
	"fmt"
	"github.com/redis/go-redis/v9"
	"log"
	"time"
)

const (
	leaderKey        = "cron:leader"
	leaderTTL        = 10 * time.Second
	leaderRetryDelay = 3 * time.Second

	taskLockPrefix = "cron:task_lock:"
	lockDuration   = 30 * time.Second

	deployMode = "standalone"
)

type Coordinator struct {
	Mode     string
	client   *redis.Client
	nodeID   string
	isLeader bool
	stopChan chan struct{}
	ctx      context.Context
}

func NewCoordinator(nodeID string, deployMode string, client *redis.Client) *Coordinator {
	return &Coordinator{
		Mode:     deployMode,
		client:   client,
		nodeID:   nodeID,
		isLeader: false,
		stopChan: make(chan struct{}),
		ctx:      context.Background(),
	}
}

// 启动选举过程（如果需要）
func (c *Coordinator) ElectLeader() (bool, error) {
	if c.Mode == deployMode {
		c.isLeader = true
		return true, nil
	}
	success, err := c.client.SetNX(c.ctx, leaderKey, c.nodeID, leaderTTL).Result()
	if err != nil {
		return false, err
	}
	c.isLeader = success
	return success, nil
}

// 维护领导权 保持Leader的活性
func (c *Coordinator) MaintainLeadership() {
	ticker := time.NewTicker(leaderTTL / 2)
	defer ticker.Stop()
	if c.Mode == deployMode {
		c.ElectLeader()
		return // 单机模式不需要维护领导权
	}
	for {
		select {
		case <-ticker.C:
			if c.isLeader {
				success, err := c.client.Expire(c.ctx, leaderKey, leaderTTL).Result()
				if err != nil || !success {
					c.isLeader = false
					//尝试自动选举新的Leader
					c.ElectLeader()
				}
			} else {
				//如果不是Leader，尝试重新选举
				c.ElectLeader()
			}
		case <-c.stopChan:
			if c.isLeader {
				// 释放领导权
				val, err := c.client.Get(c.ctx, leaderKey).Result()
				if err == nil && val == c.nodeID {
					c.client.Del(c.ctx, leaderKey)
				}
			}
			return
		}
	}
}
func (rc *Coordinator) Stop() {
	close(rc.stopChan)
}

func (rc *Coordinator) IsLeader() bool {
	return rc.isLeader
}

// 尝试获取任务锁
func (rc *Coordinator) TryLockTask(taskID int) (bool, error) {
	if rc.Mode == deployMode {
		return true, nil // 单机模式不需要锁
	}
	key := fmt.Sprintf("%s%d", taskLockPrefix, taskID)
	// 使用SETNX实现锁
	success, err := rc.client.SetNX(rc.ctx, key, rc.nodeID, lockDuration).Result()
	if err != nil {
		return false, err
	}
	return success, nil
}

// UnlockTask 释放任务锁
func (rc *Coordinator) UnlockTask(taskID int) error {
	if rc.Mode == deployMode {
		return nil // 单机模式不需要释放锁
	}
	key := fmt.Sprintf("%s%d", taskLockPrefix, taskID)

	// 使用Lua脚本确保只有锁的持有者才能释放
	script := `
	if redis.call("GET", KEYS[1]) == ARGV[1] then
		return redis.call("DEL", KEYS[1])
	else
		return 0
	end
	`
	_, err := rc.client.Eval(rc.ctx, script, []string{key}, rc.nodeID).Result()
	return err
}

const (
	taskUpdateChannel = "cron:task_updates"
)

// PublishTaskUpdate 发布任务更新通知
func (rc *Coordinator) PublishTaskUpdate(taskID int, action string) error {
	if rc.Mode == deployMode {
		return nil // 单机模式不需要发布通知
	}
	message := fmt.Sprintf("%d:%s", taskID, action)
	return rc.client.Publish(rc.ctx, taskUpdateChannel, message).Err()
}

// SubscribeTaskUpdates 订阅任务更新通知
func (rc *Coordinator) SubscribeTaskUpdates(handler func(taskID int, action string)) {
	if rc.Mode == deployMode {
		return // 单机模式不需要订阅
	}
	pubsub := rc.client.Subscribe(rc.ctx, taskUpdateChannel)
	go func() {
		ch := pubsub.Channel()
		for msg := range ch {
			var taskID int
			var action string
			_, err := fmt.Sscanf(msg.Payload, "%d:%s", &taskID, &action)
			if err != nil {
				log.Printf("Invalid task update message: %s", msg.Payload)
				continue
			}
			handler(taskID, action)
		}
	}()
}
